package com.example.mq.mqserver;

import com.example.mq.common.Consumer;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;

import javax.crypto.MacSpi;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 虚拟主机
 * 每个虚拟主机下面都管理自己的 交换机、队列、绑定、消息、数据
 * 同时提供 api 供上层使用
 */
public class VirtualHost {

    private String virtualHostName;

    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();

    private DiskDataCenter diskDataCenter = new DiskDataCenter();

    private Router router = new Router();

    private ConsumerManager consumerManager = new ConsumerManager(this);

    // 操作交换机的锁对象
    private final Object exchangeLocker = new Object();

    // 操作队列的锁对象
    private final Object queueLocker = new Object();

    public VirtualHost(String name){
        this.virtualHostName = name;

        // 对于 MemoryDataCenter 来说，不需要额外的初始化操作，只要对象 new 出来就行了
        // 但是，DiskDataCenter 需要初始化
        diskDataCenter.init();

        // 还需要针对硬盘的数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内容数据失败！");
        }
    }

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    // 创建交换机
    // 如果交换机不存在就创建，存在就返回
    // 返回值为 boolean
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments){
        // 将交换机的名字加上虚拟主机的名字作为前缀
        exchangeName = virtualHostName + exchangeName;
        // 1. 通过内存查询，判定交换机是否存在
        try{
            synchronized (exchangeLocker){
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if(null != existsExchange){
                    // 交换机存在
                    System.out.println("[VirtualHost] 交换机已经存在！ exchangeName = " + exchangeName);
                    return true;
                }
                // 2. 真正创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //先写硬盘后写内存，因为硬盘更容易写数百，如果硬盘失败内存就不写了
                //要是先写内存，内存成功但是硬盘失败，还要将内存中的数据删掉
                // 3. 将交换机对象写入硬盘
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                // 4. 把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功！ exchangeName = " + exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败！ exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                // 1. 先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if(null == toDelete){
                    throw new MqException("[VirtualHost] 交换机不存在无法删除！");
                }
                // 2. 删除硬盘上得到数据
                if(toDelete.isDurable()){
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3. 删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！ exchangeName = " + exchangeName);

            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败！ exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    // 创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments){
        // 将队列的名字加上虚拟主机的名字作为前缀
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                // 1. 判定队列是否存在
                MSGQueue existQueue = memoryDataCenter.getQueue(queueName);
                if(null != existQueue){
                    System.out.println("[VirtualHost] 队列已经存在！ queueName = " + queueName);
                    return true;
                }
                // 2. 创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(true);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                // 3. 写硬盘
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                // 4. 写内存
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！ queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列创建失败！ queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除队列
    public boolean queueDelete(String queueName){
        queueName = virtualHostName + queueName;
        try{
            synchronized (queueLocker){
                // 1. 根据队列名字，查询当前队列是否存在
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(null == queue){
                    throw new MqException("[VirtualHost] 队列不存在！ queueName = " + queueName);
                }
                // 2. 删除硬盘数据
                if(queue.isDurable()){
                    diskDataCenter.deleteQueue(queueName);
                }
                // 3. 删除内存数据
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功！ queueName = " + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列删除失败！ queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 将一个队列和一个交换机绑定
    public boolean queueBinding(String queueName, String exchangeName, String bindingKey){
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    // 1. 判定当前的绑定是否存在
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(null != existsBinding){
                        throw new MqException("[virtualHost] binding 已经存在！ exchangeName = " + exchangeName +
                                ", queueName = " + queueName);
                    }
                    // 2. 验证 bindingKey 是否合法
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey 非法！ bindingKey = " + bindingKey);
                    }
                    // 3. 创建 Binding 对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    // 4. 获取一下对应交换的和队列，如果不存在是无法创建的
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(null == exchange){
                        throw new MqException("[VirtualHost] 交换机不存在！ exchangeName = " + exchangeName);
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(null == queue){
                        throw new MqException("[VirtualHost] 队列不存在！ queueName = " + queueName);
                    }
                    // 5. 先写硬盘
                    if(exchange.isDurable() && queue.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    // 6. 再写内存
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功！ exchangeName = " + exchangeName +
                            ", queueName = " + queueName);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败！ exchangeName = " + exchangeName +
                    ", queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 删除绑定
    public boolean queueUnBinding(String exchangeName, String queueName){
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try{
           synchronized (exchangeLocker){
               synchronized (queueLocker){
                   // 1. 获取 binding 看是否已经存在
                   Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                   if(null == binding){
                       throw new MqException("[VirtulaHost] 删除绑定失败！ 绑定不存在! exchangeName = " + exchangeName +
                               ", queueName = " + queueName);
                   }
                   diskDataCenter.deleteBinding(binding);
                   // 2. 删除内存上的数据
                   memoryDataCenter.deleteBinding(binding);
                   System.out.println("[VirtualHost] 删除绑定成功！ exchangeName = " + exchangeName +
                           ", queueName = " + queueName);
               }
           }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败！ exchangeName = " + exchangeName +
                    ", queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    // 发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body){
        try{
            // 1. 转换交换机的名字
            exchangeName = virtualHostName + exchangeName;
            // 2. 检查 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey 非法！ routingKey = " + routingKey);
            }
            // 3. 查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(null == exchange){
                throw new MqException("[VirtualHost] 交换机不存在！ exchangeName = " + exchangeName);
            }
            // 4. 判定交换机的类型
            if(exchange.getType() == ExchangeType.DIRECT){
                // 按照直接交换机的方式来转发消息
                // 以 routingKey 作为队列的名字，直接把消息写入指定的队列中
                // 此时，可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                // 5. 构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                // 6. 查找该队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(null == queue){
                    throw new MqException("[VirtualHost] 队列不存在！ queueName = " + queueName);
                }
                // 7. 队列存在，直接往队列中写入消息
                sendMesssage(queue, message);
            }else{
                // 按照 fanout 和 topic 的方式来转发
                // 5. 找到该交换机关联的所有绑定， 并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for(Map.Entry<String, Binding> entry : bindingsMap.entrySet()){
                    // 1) 获取到绑定对象，判定对应的队列是否存在
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(null == queue){
                        System.out.println("[VirtualHost] basicPublish 发送消息时，发现队列不存在！ queueName = " + binding.getQueueName());
                        continue;
                    }
                    // 2) 构造消息对象
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    // 3) 判定这个消息是否能转发给该队列
                    //    如果为 fanout ，所有绑定的队列都要转发
                    //    如果为 topic ，还需要判定 bindingKey 和 routingKey 是否匹配
                    if(!router.route(exchange.getType(), binding, message)){
                        continue;
                    }
                    // 4) 真正转发消息给队列
                    sendMesssage(queue, message);
                }
            }
            System.out.println("[VirtualHost] 消息发送成功！");
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 消息发送失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMesssage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 将消息写入到 硬盘 和 内存 上
        int deliverMode = message.getDeliverMode();
        // 写入硬盘， deliverMode = 1 为 不持久化，deliverMode = 2 为 持久化
        if(deliverMode == 2){
            diskDataCenter.sendMessage(queue, message);
        }
        // 写入内存
        memoryDataCenter.sendMessage(queue, message);

        // 通知告知消费者可以消费消息了
        consumerManager.notifyConsume(queue.getName());
    }

    /**
     * 订阅消息
     * 添加一个队列的订阅者，当队列收到消息后，就要把消息推送给对应的订阅者
     * @param consumerTag 消费者的身份标识
     * @param queueName
     * @param autoAck 消息被消费完成后，应答的方式，true 自动应答，false 手动应答
     * @param consumer 是一个回调函数，类型设定成函数式接口，后续调用 basicConsume 并且传实参的时候，可以写成 lambda 的样子
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer){
        // 构造一个 ConsumerEnv 对象，把这个对象对应的队列找到，再把这个 Consumer 对象添加到该队列中
        queueName = virtualHostName + queueName;
        try{
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            System.out.println("[VirtualHost] basicConsume 成功！ queueName = " + queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicConsume 失败！ queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try{
            // 1. 获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if(null == message){
                throw new MqException("[VirtualHost] 要确认的消息不存在！ messageId = " + messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(null == queue){
                throw new MqException("[VirtualHost] 要确认的队列不存在! queueName = " + queueName);
            }
            // 2. 删除硬盘上的数据
            if(2 == message.getDeliverMode()){
                diskDataCenter.deleteMessage(queue,message);
            }
            // 3. 删除内存的数据
            memoryDataCenter.removeMessage(messageId);
            // 4. 删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName, messageId);
            System.out.println("[VirtualHost] basicAck 成功！消息被成功确认！ queueName = " + queueName
                    + ", messageId = " + messageId);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicAck 失败！消息确认失败！ queueName = " + queueName
                    + ", messageId = " + messageId);
            e.printStackTrace();
            return false;
        }
    }
}
